package com.captjack.bigdata.hadoop.hdfs.impl;

import com.captjack.bigdata.hadoop.hdfs.HadoopDistributedFileSystemDao;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileAlreadyExistsException;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Repository;

import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;

/**
 * @author Capt Jack
 * @date 2017/10/24
 * 大数据文件系统操作工具
 */
@Repository
public class HadoopDistributedFileSystemDaoImpl implements HadoopDistributedFileSystemDao {

    private static final Logger logger = LoggerFactory.getLogger(HadoopDistributedFileSystemDaoImpl.class);

    /**
     *
     */
    private static final int BUFFER_SIZE = 4096;

    /**
     * 大数据配置
     */
    private Configuration configuration;

    /**
     * 获取hadoop文件系统
     *
     * @param path          需要获取的路径
     * @param configuration hadoopFileSystem配置
     * @return FileSystem
     * @throws IOException 异常
     */
    private FileSystem getFileSystem(String path, Configuration configuration) throws IOException {
        try {
            if (StringUtils.isBlank(path)) {
                throw new IllegalArgumentException("The path is unavailable！path : " + path);
            }
            return FileSystem.get(URI.create(path), configuration);
        } catch (IOException ioe) {
            logger.error("", ioe);
            throw ioe;
        }
    }

    /**
     * 创建目录
     *
     * @param directoryPath 目录路径
     * @return 返回true代表创建成功，反之则创建失败
     * @throws IOException 异常
     */
    @Override
    public boolean createDirectory(String directoryPath) throws IOException {
        // 获取文件系统
        FileSystem fileSystem = null;
        try {
            fileSystem = getFileSystem(directoryPath, configuration);
            // 路径
            Path path = new Path(directoryPath);
            // 如果目录不存在，那么新建目录
            if (!fileSystem.exists(path)) {
                fileSystem.mkdirs(path);
            } else {
                throw new FileAlreadyExistsException("Directory already exist！path : " + directoryPath);
            }
            return true;
        } catch (IOException ioe) {
            logger.error("", ioe);
            throw ioe;
        } finally {
            // 关闭流
            if (fileSystem != null) {
                fileSystem.close();
            }
        }
    }

    /**
     * 创建新文件
     *
     * @param filePath 目录路径
     * @return 返回true代表创建成功，反之则创建失败
     * @throws IOException 异常
     */
    @Override
    public boolean createFile(String filePath) throws IOException {
        // 获取文件系统
        FileSystem fileSystem = null;
        // fs输入流
        FSDataOutputStream fsDataOutputStream = null;
        try {
            fileSystem = getFileSystem(filePath, configuration);
            // 如果文件不存在，那么新建文件
            Path path = new Path(filePath);
            if (!fileSystem.exists(path)) {
                fsDataOutputStream = fileSystem.create(path);
                return true;
            } else {
                fileSystem.close();
                throw new FileAlreadyExistsException("File already exist！path : " + filePath);
            }
        } catch (IOException ioe) {
            logger.error("", ioe);
            throw ioe;
        } finally {
            // 关闭流
            if (fsDataOutputStream != null) {
                fsDataOutputStream.close();
            }
            if (fileSystem != null) {
                fileSystem.close();
            }
        }
    }

    /**
     * 向文件中写入内容，如果文件不存在，那么创建新文件写入
     *
     * @param filePath    文件路径
     * @param content     写入内容，字节数组
     * @param isOverWrite 是否覆盖，true为覆盖，反之则是追加
     * @return 返回true，代表写入成功，反之则是失败
     * @throws IOException 异常
     */
    @Override
    public boolean write(String filePath, byte[] content, boolean isOverWrite) throws IOException {
        if (content == null || content.length == 0 || filePath == null) {
            logger.error("Write content is null or filePath is null！");
            throw new IllegalArgumentException("Write content is empty！");
        }
        FileSystem fileSystem = null;
        FSDataOutputStream fsDataOutputStream = null;
        try {
            // 获取文件系统
            fileSystem = getFileSystem(filePath, configuration);
            // 路径
            Path path = new Path(filePath);
            // 如果文件不存在，那么新建文件，或者是覆盖模式，那么直接写入文件
            if (!fileSystem.exists(path) || isOverWrite) {
                fsDataOutputStream = fileSystem.create(path, true);
                fsDataOutputStream.write(content);
            } else {
                // 复制流到文件
                IOUtils.copyBytes(new ByteArrayInputStream(content), fileSystem.append(path), BUFFER_SIZE, true);
            }
            return true;
        } catch (IOException ioe) {
            logger.error("write file failed! filePath : " + filePath);
            throw ioe;
        } finally {
            if (fsDataOutputStream != null) {
                fsDataOutputStream.close();
            }
            if (fileSystem != null) {
                fileSystem.close();
            }
        }
    }

    /**
     * 向文件中写入内容，如果文件不存在，那么创建新文件写入
     *
     * @param filePath    文件路径
     * @param file        写入文件
     * @param isOverWrite 是否覆盖，true为覆盖，反之则是追加
     * @return 返回true，代表写入成功，反之则是失败
     * @throws IOException 异常
     */
    @Override
    public boolean write(String filePath, File file, boolean isOverWrite) throws IOException {
        try {
            InputStream inputStream = new FileInputStream(file);
            byte[] bytes = new byte[inputStream.available()];
            inputStream.read(bytes);
            return write(filePath, bytes, isOverWrite);
        } catch (IOException ioe) {
            logger.error("", ioe);
            throw ioe;
        }
    }

    /**
     * 读取hdfs文件
     *
     * @param filePath 文件路径
     * @return 字节数组
     * @throws IOException 异常
     */
    @Override
    public byte[] read(String filePath) throws IOException {
        if (filePath == null) {
            logger.error("read hdfs failed because filePath is null！");
            throw new IllegalArgumentException("filePath is null！");
        }
        FileSystem fileSystem = null;
        FSDataInputStream fsDataInputStream = null;
        try {
            // 获取文件系统
            fileSystem = getFileSystem(filePath, configuration);
            // 文件是否存在
            Path path = new Path(filePath);
            if (fileSystem.exists(path)) {
                // 打开输入流
                fsDataInputStream = fileSystem.open(path);
                // 获取文件状态
                FileStatus fileStatus = fileSystem.getFileStatus(path);
                // 创建字节数组
                byte[] buffer = new byte[Integer.parseInt(fileStatus.getLen() + "")];
                // 一次性读取所有内容
                fsDataInputStream.readFully(0, buffer);
                return buffer;
            } else {
                logger.error("The file is not found！path : " + filePath);
                throw new FileNotFoundException("The file is not found！path : " + filePath);
            }
        } catch (IOException ioe) {
            logger.error("", ioe);
            throw ioe;
        } finally {
            if (fsDataInputStream != null) {
                fsDataInputStream.close();
            }
            if (fileSystem != null) {
                fileSystem.close();
            }
        }
    }

    /**
     * 删除文件或者目录
     *
     * @param filePath 文件路径
     * @return true代表成功，反之代表失败
     * @throws IOException 异常
     */
    @Override
    public boolean deleteHDFSFile(String filePath) throws IOException {
        // 获取文件系统
        FileSystem fileSystem = null;
        try {
            // 获取文件系统
            fileSystem = getFileSystem(filePath, configuration);
            // 创建路径
            Path path = new Path(filePath);
            // 删除文件，执行操作，返回结果
            return fileSystem.delete(path, true);
        } catch (IOException ioe) {
            logger.error("", ioe);
            throw ioe;
        } finally {
            if (fileSystem != null) {
                fileSystem.close();
            }
        }
    }

    /**
     * 列出文件夹下所有文件，不递归
     *
     * @param path 路径
     * @return 文件名集合
     * @throws IOException 异常
     */
    @Override
    public List<String> listAll(String path) throws IOException {
        FileSystem fileSystem = null;
        try {
            // 获取文件系统
            fileSystem = getFileSystem(path, configuration);
            // 获取文件状态数组
            FileStatus[] fileStatuses = fileSystem.listStatus(new Path(path));
            // 返回
            List<String> fileList = new ArrayList<String>();
            for (FileStatus fileStatus : fileStatuses) {
                fileList.add(fileStatus.getPath().toString());
            }
            return fileList;
        } catch (IOException ioe) {
            logger.error("", ioe);
            throw ioe;
        } finally {
            if (fileSystem != null) {
                fileSystem.close();
            }
        }
    }

    /**
     * 创建hadoop统一配置
     *
     * @param configuration hadoop配置
     */
    @Autowired
    @Qualifier("hadoopFileSystemConfiguration")
    public void setConfiguration(Configuration configuration) {
        this.configuration = configuration;
        // 每次创建新的FileSystem，不从缓存中获取
        configuration.set("fs.hdfs.impl.disable.cache", "true");
    }

}